package com.atguigu.flink.sql.query;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * Created by 黄凯 on 2023/6/27 0027 21:26
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 *
 * 用户自定义函数
 *  *  1. 标量函数 （一进一出）
 *  *  2. 表函数   (一进多出)
 *  *  3. 聚合函数  (多进一出)
 *  *  4. 表聚合函数 (多进多出)
 */
public class Flink13_userDefineFunction {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("127.0.0.1", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new WaterSensor(fields[0].trim(), Long.valueOf(fields[1].trim()), Integer.valueOf(fields[2].trim()));
                        }

                );
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Schema schema = Schema.newBuilder()
                .column("id", "string")
                .column("vc" , "int")
                .column("ts" , "bigint")
                .columnByExpression("pt" , "proctime()")
                .columnByExpression("et" , "to_timestamp_ltz(ts, 3)")
                .watermark("et" , "et - interval '0' second")
                .build();
        Table table = tableEnv.fromDataStream(ds, schema);
        tableEnv.createTemporaryView("t1" , table);

        //使用自定义标量函数
        //注册函数
//        tableEnv.createTemporaryFunction("myupper" , MyUpperFunction.class);
//        API
//        table.select(call("myupper" , $("id")) , $("vc") , $("ts")).execute().print();
//        SQL
//        tableEnv.sqlQuery("select myupper(id) , vc, ts from t1").execute().print();


        //使用自定义表函数
//        tableEnv.createTemporaryFunction("mysplit" , MySplitFunction.class);
        //API
        /**
         * +----+--------------------------------+--------------------------------+-------------+-------------+----------------------+
         * | op |                             id |                           word |      length |          vc |                   ts |
         * +----+--------------------------------+--------------------------------+-------------+-------------+----------------------+
         * | +I |                             s1 |                             s1 |           2 |           5 |                 1000 |
         * | +I |                       s1-s2-s3 |                             s1 |           2 |           5 |                 1000 |
         * | +I |                       s1-s2-s3 |                             s2 |           2 |           5 |                 1000 |
         * | +I |                       s1-s2-s3 |                             s3 |           2 |           5 |                 1000 |
         */
//        tableEnv
//                .from("t1")
//                .joinLateral(call("mysplit", $("id")))
//                .select($("id"), $("word"), $("length") , $("vc") , $("ts"))
//                .execute()
//               .print();
        //SQL
//        tableEnv.sqlQuery(
//                "SELECT id,  newWord, newLength  , vc , ts " +
//                        "FROM t1 " +
//                        "LEFT JOIN LATERAL TABLE(mysplit(id)) AS T(newWord, newLength) ON TRUE")
//                .execute().print();

        //使用自定义聚合函数
//        tableEnv.createTemporaryFunction("myavg" , MyAvgFunction.class);
        //API
//        table.groupBy($("id")).select($("id") , call("myavg", $("vc")))
//               .execute().print();

        //SQL
//        tableEnv.sqlQuery("select id , myavg(vc) from t1 group by id ").execute().print();

        //使用表聚合函数
        tableEnv.createTemporaryFunction("mytop2" , MyTop2Function.class);
        //API
        /**
         * +----+--------------------------------+-------------+-------------+
         * | op |                             id |           v |        rank |
         * +----+--------------------------------+-------------+-------------+
         * | +I |                              1 |           5 |           1 |
         * | -D |                              1 |           5 |           1 |
         * | +I |                              1 |           6 |           1 |
         * | +I |                              1 |           5 |           2 |
         * | -D |                              1 |           6 |           1 |
         * | -D |                              1 |           5 |           2 |
         * | +I |                              1 |           7 |           1 |
         * | +I |                              1 |           6 |           2 |
         */
        table.groupBy($("id"))
                .flatAggregate(call("mytop2" , $("vc")) )
                .select($("id") , $("v")  , $("rank"))
                .execute()
                .print();


    }

    /**
     * 自定义标量函数
     * 1. 自定义类继承ScalarFunction
     * 2. 实现eval方法 , 强制要求叫eval
     */

    public static class MyUpperFunction extends ScalarFunction {

        public String eval( String input) {
            return input.toUpperCase() ;
        }
    }

    /**
     * 自定义表函数
     *
     *  water-sensor-1  => split("water-sensor-1") =>
     *                                                water  5
     *                                                sensor 6
     *                                                1      1
     */

    @FunctionHint(output =  @DataTypeHint("ROW<word STRING , length INT>"))
    public static class MySplitFunction extends TableFunction<Row> {
        public void eval(String str) {
            String[] words = str.split("-");
            for (String word : words) {
                collect(Row.of( word, word.length()));
            }
        }
    }

    /**
     * 自定义聚合函数
     * 计算每种传感器的平均水位记录
     */

    public static class MyAvgFunction extends AggregateFunction<Double , Tuple2<Long , Long>> {

        @Override
        public Double getValue(Tuple2<Long, Long> acc) {
            return (double)acc.f0 / acc.f1;
        }

        @Override
        public Tuple2<Long, Long> createAccumulator() {
            return Tuple2.of(0L, 0L);
        }

        public void accumulate(Tuple2<Long,Long> acc, Integer vc ) {
            acc.f0 +=vc ;
            acc.f1 +=1;
        }
    }

    /**
     * 自定义表聚合函数
     *
     * 求每种传感器水位记录的前两名
     */
    @FunctionHint(output = @DataTypeHint("ROW<v Integer , rank Integer>"))
    public static class MyTop2Function extends TableAggregateFunction<Row , Tuple2<Integer ,Integer>> {

        @Override
        public Tuple2<Integer, Integer> createAccumulator() {
            return Tuple2.of( Integer.MIN_VALUE , Integer.MIN_VALUE );
        }

        public void accumulate(Tuple2<Integer ,Integer> acc, Integer v) {
            //判断新数据是否大于第二名
            if( v > acc.f1){
                //判断是否大于第一名
                if(v > acc.f0){
                    //第一名给第二名
                    acc.f1 = acc.f0 ;
                    //新值给第一名
                    acc.f0 = v ;
                }else{
                    //新值给第二名
                    acc.f1 = v ;
                }
            }
        }
        public void emitValue(Tuple2<Integer ,Integer> acc, Collector<Row> out) {
            //发射第一名
            out.collect(Row.of(acc.f0 , 1));
            //判断是否有第二名
            if(acc.f1 != Integer.MIN_VALUE){
                //发射第二名
                out.collect(Row.of(acc.f1 , 2));
            }
        }
    }

}
